package com.seal.commons.infrastructure.redis.command;

import com.seal.commons.infrastructure.facade.cache.RedisCommand;
import com.seal.commons.infrastructure.redis.serialize.RedisSerializer;
import io.lettuce.core.KeyValue;
import io.lettuce.core.api.StatefulConnection;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import io.lettuce.core.cluster.api.sync.RedisClusterCommands;
import org.springframework.lang.Nullable;

import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * 功能描述: <br/>
 *
 * @author yong.huang
 * @date: 2020-07-16 21:59<br/>
 * @since JDK 1.8
 */
public class DefaultRedisCommand<K, V> implements RedisCommand<K, V> {

    private RedisSerializer keySerializer ;

    private RedisSerializer valueSerializer;

    private RedisSerializer hashKeySerializer;

    private StatefulConnection connection;

    public DefaultRedisCommand(StatefulConnection connection) {
        this.connection = connection;
    }

    private RedisClusterCommands<byte[], byte[]> getConnection(){
        if (connection instanceof StatefulRedisConnection) {
            return ((StatefulRedisConnection<byte[], byte[]>) connection).sync();
        }
        if (connection instanceof StatefulRedisClusterConnection) {
            return ((StatefulRedisClusterConnection<byte[], byte[]>) connection).sync();
        }
        throw new IllegalStateException(
                String.format("%s is not a supported connection type.", connection.getClass().getName()));
    }

    @Override
    public Boolean expire(K key, long expire) {
        byte[] bytes = getKeySerializer().serialize(key);
        return  getConnection().expire(bytes, expire);
    }

    @Override
    public Boolean expire(K key, long expire, TimeUnit unit) {
        byte[] bytes = getKeySerializer().serialize(key);
        return getConnection().expire(bytes, TimeUnit.SECONDS.convert(expire, unit));
    }

    @Override
    public Boolean persist(K key) {
        byte[] bytes = getKeySerializer().serialize(key);
        return getConnection().persist(bytes);
    }

    @Override
    public Long getExpire(K key, TimeUnit unit) {
        byte[] bytes = getKeySerializer().serialize(key);
        Long expire =  getConnection().ttl(bytes);
        if(unit == null){
            return expire;
        }
        return unit.convert(expire, TimeUnit.SECONDS);
    }

    @Override
    public void put(K key, V value) {
        byte[] keyBytes = getKeySerializer().serialize(key);
        byte[] valueBytes = getValueSerializer().serialize(value);
        getConnection().set(keyBytes, valueBytes);
    }

    @Override
    public void put(K key, V value, long expire) {
        byte[] keyBytes = getKeySerializer().serialize(key);
        byte[] valueBytes = getValueSerializer().serialize(value);
        getConnection().setex(keyBytes, expire, valueBytes);
    }

    @Override
    public void put(K key, V value, long expire, TimeUnit unit) {
        byte[] keyBytes = getKeySerializer().serialize(key);
        byte[] valueBytes = getValueSerializer().serialize(value);
        getConnection().setex(keyBytes, unit.toMillis(expire), valueBytes);
    }

    @Override
    public Boolean del(K key) {
        byte[] keyBytes = getKeySerializer().serialize(key);
        Long result =  getConnection().del(keyBytes);
        return result != null && result.intValue() == 1;
    }

    @Override
    public Boolean exists(K key) {
        byte[] keyBytes = getKeySerializer().serialize(key);
        Long count =  getConnection().exists(keyBytes);
        return count != null ? count > 0 : null;
    }

    @Override
    public V get(K key) {
        byte[] keyBytes = getKeySerializer().serialize(key);
        byte[] valueBytes = getConnection().get(keyBytes);
        return (V)getValueSerializer().deserialize(valueBytes);
    }

    @Override
    public Long incr(K  key) {
        byte[] keyBytes = getKeySerializer().serialize(key);
        return getConnection().incr(keyBytes);
    }

    @Override
    public Long incr(K key, Long increment) {
        byte[] keyBytes = getKeySerializer().serialize(key);
        return getConnection().incrby(keyBytes, increment);
    }

    @Override
    public <HK, HV> HV hGet(K key, HK hashKey) {
        byte[] keyBytes = getKeySerializer().serialize(key);
        byte[] hashKeyBytes = getHashKeySerializer().serialize(hashKey);
        byte[] valueByte =  getConnection().hget(keyBytes, hashKeyBytes);
        return (HV)getValueSerializer().deserialize(valueByte);
    }

    @Override
    public <HK> Boolean hDel(K key, HK hashKey) {
        byte[] keyBytes = getKeySerializer().serialize(key);
        byte[] hashKeyBytes = getHashKeySerializer().serialize(hashKey);
        Long result =  getConnection().hdel(keyBytes, hashKeyBytes);
        return result != null && result.intValue() == 1;
    }

    @Override
    public <HK, HV> Map<HK, HV> hGetAll(K key) {
        byte[] keyBytes = getKeySerializer().serialize(key);
        Map<byte[], byte[]> entries =  getConnection().hgetall(keyBytes);
        return deserializeHashMap(entries);
    }


    <HK, HV> Map<HK, HV> deserializeHashMap(@Nullable Map<byte[], byte[]> entries) {
        // connection in pipeline/multi mode

        if (entries == null) {
            return null;
        }
        Map<HK, HV> map = new LinkedHashMap<>(entries.size());

        for (Map.Entry<byte[], byte[]> entry : entries.entrySet()) {
            map.put((HK) getHashKeySerializer().deserialize(entry.getKey()), (HV) getValueSerializer().deserialize(entry.getValue()));
        }

        return map;
    }


    @Override
    public <HK, HV>  List<HV> hMultiGet(K key, Collection<HK> hashKeys) {
        if (hashKeys.isEmpty()) {
            return Collections.emptyList();
        }
        byte[] rawKey = getKeySerializer().serialize(key);
        byte[][] rawHashKeys = new byte[hashKeys.size()][];
        int counter = 0;
        for (Object hashKey : hashKeys) {
            rawHashKeys[counter++] = getHashKeySerializer().serialize(hashKey);
        }
        List<KeyValue<byte[], byte[]>> values = getConnection().hmget(rawKey, rawHashKeys);
        List<HV> list = new ArrayList<>(values.size());
        for(KeyValue<byte[], byte[]> value : values){
            list.add((HV)getValueSerializer().deserialize(value.getValue()));
        }
        return list;
    }




    @Override
    public  <HK, HV> void hSet(K key, HK hashKey, HV value) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[] hashKeyByte = getHashKeySerializer().serialize(hashKey);
        byte[] valueByte = getValueSerializer().serialize(value);
        getConnection().hset(keyByte, hashKeyByte, valueByte);
    }

    @Override
    public <HK, HV> void hSetAll(K key, Map<HK, HV> map) {
        byte[] keyByte = getKeySerializer().serialize(key);
        Map<byte[] , byte[]> entries = new HashMap<>();
        for(Map.Entry<HK, HV> entry : map.entrySet()){
            entries.put(getHashKeySerializer().serialize(entry.getKey()), getValueSerializer().serialize(entry.getValue()));
        }
        getConnection().hmset(keyByte, entries);
    }

    @Override
    public <HK> boolean hExists(K key, HK hashKey) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[] hashKeyByte = getHashKeySerializer().serialize(hashKey);
        return getConnection().hexists(keyByte, hashKeyByte);
    }

    @Override
    public <V> V lIndex(K key, long index) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[] objectValues =  getConnection().lindex(keyByte, index);
        return (V)getValueSerializer().deserialize(objectValues);
    }

    @Override
    public <V> List<V> lRange(K key, long start, long end) {
        byte[] keyByte = getKeySerializer().serialize(key);
        List<byte[]> objectValues =  getConnection().lrange(keyByte, start, end);
        if(objectValues.isEmpty()){
            return null;
        }
        List<V> list = new LinkedList<>();
        for(byte[] bytes : objectValues){
            list.add((V)getValueSerializer().deserialize(bytes));
        }
        return list;
    }

    @Override
    public Long lLeftPush(K key, V value) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[] valueByte = getValueSerializer().serialize(value);
        return getConnection().lpush(keyByte, valueByte);
    }



    @Override
    public Long lLeftPushAll(K key, V... values) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[][] valueBytes = new byte[values.length][];
        int count = 0;
        for(V value : values){
            valueBytes[count++] = getValueSerializer().serialize(value);
        }
        return getConnection().lpush(keyByte, valueBytes);
    }

    @Override
    public Long lLeftPushAll(K key, Collection<V> values) {
        if(values.isEmpty()){
            return null;
        }
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[][] valueBytes = new byte[values.size()][];
        int count = 0;
        for(Object value : values){
            valueBytes[count++] = getValueSerializer().serialize(value);
        }
        return getConnection().lpush(keyByte, valueBytes);
    }

    @Override
    public Long lLeftPushIfPresent(K key, V value) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[] valueByte = getValueSerializer().serialize(value);
        return getConnection().lpushx(keyByte, valueByte);
    }

    @Override
    public Long lRightPush(K key, V value) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[] valueByte = getValueSerializer().serialize(value);
        return getConnection().rpush(keyByte, valueByte);
    }

    @Override
    public Long lRightPushAll(K key, V... values) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[][] valueBytes = new byte[values.length][];
        int count = 0;
        for(V value : values){
            valueBytes[count++] = getValueSerializer().serialize(value);
        }
        return getConnection().rpush(keyByte, valueBytes);
    }

    @Override
    public Long lRightPushAll(K key, Collection<V> values) {
        if(values.isEmpty()){
            return null;
        }
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[][] valueBytes = new byte[values.size()][];
        int count = 0;
        for(V value : values){
            valueBytes[count++] = getValueSerializer().serialize(value);
        }
        return getConnection().rpush(keyByte, valueBytes);
    }

    @Override
    public Long lRightPushIfPresent(K key, V value) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[] valueByte = getValueSerializer().serialize(value);
        return getConnection().rpushx(keyByte, valueByte);
    }


    @Override
    public void lSet(K key, long index, V value) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[] valueByte = getValueSerializer().serialize(value);
        getConnection().lset(keyByte, index, valueByte);
    }

    @Override
    public <V> V lRightPop(K key) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[] valueByte = getConnection().rpop(keyByte);
        return (V)getValueSerializer().deserialize(valueByte);
    }

    @Override
    public <V> V lLeftPop(K key) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[] valueByte = getConnection().lpop(keyByte);
        return (V)getValueSerializer().deserialize(valueByte);
    }

    @Override
    public Long sAdd(K key, V... values) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[][] valueBytes = new byte[values.length][];
        int count = 0;
        for(Object value : values){
            valueBytes[count++] = getValueSerializer().serialize(value);
        }
        return getConnection().sadd(keyByte, valueBytes);
    }

    @Override
    public Long sRemove(K key, V... values) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[][] valueBytes = new byte[values.length][];
        int count = 0;
        for(V value : values){
            valueBytes[count++] = getValueSerializer().serialize(value);
        }
        return getConnection().srem(keyByte, valueBytes);
    }

    @Override
    public <V> V sRandomMember(K key) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[] valueByte = getConnection().srandmember(keyByte);
        return (V)getValueSerializer().deserialize(valueByte);

    }

    @Override
    public <V> List<V> sRandomMembers(K key, long count) {
        byte[] keyByte = getKeySerializer().serialize(key);
        List<byte[]> valueBytes = getConnection().srandmember(keyByte, count);
        if(valueBytes.isEmpty()){
            return null;
        }
        List<V> list = new LinkedList<>();
        for(byte[] bytes : valueBytes){
            list.add((V)getValueSerializer().deserialize(bytes));
        }
        return list;
    }

    @Override
    public Boolean sMove(K key, V value, K destKey) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[] valueByte = getValueSerializer().serialize(value);
        byte[] destKeyByte = getKeySerializer().serialize(destKey);
        return getConnection().smove(keyByte, destKeyByte, valueByte);
    }

    @Override
    public Boolean sIsMember(K key, V value) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[] valueByte = getValueSerializer().serialize(value);
        return getConnection().sismember(keyByte, valueByte);
    }

    @Override
    public <V> Set<V> sIntersect(K key, K otherKey) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[] otherKeyByte = getKeySerializer().serialize(otherKey);
        Set<byte[]> intersectValues = getConnection().sinter(keyByte, otherKeyByte);
        if(intersectValues.isEmpty()){
            return null;
        }
        Set<V> set = new HashSet<>();
        for(byte[] valueByte : intersectValues){
            set.add((V)getValueSerializer().deserialize(valueByte));
        }
        return set;
    }

    @Override
    public Long sIntersectAndStore(K key, K otherKey, K destKey) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[] otherKeyByte = getKeySerializer().serialize(otherKey);
        byte[] destKeyByte = getKeySerializer().serialize(destKey);
        return getConnection().sinterstore(destKeyByte, keyByte, otherKeyByte);
    }

    @Override
    public <V> Set<V> sUnion(K key, K otherKey) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[] otherKeyByte = getKeySerializer().serialize(otherKey);
        Set<byte[]> intersectValues = getConnection().sunion(keyByte, otherKeyByte);
        if(intersectValues.isEmpty()){
            return null;
        }
        Set<V> set = new HashSet<>();
        for(byte[] valueByte : intersectValues){
            set.add((V)getValueSerializer().deserialize(valueByte));
        }
        return set;
    }

    @Override
    public Long sUnionAndStore(K key, K otherKey, K destKey) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[] otherKeyByte = getKeySerializer().serialize(otherKey);
        byte[] destKeyByte = getKeySerializer().serialize(destKey);
        return getConnection().sunionstore(destKeyByte, keyByte, otherKeyByte);
    }

    @Override
    public <V> Set<V> sDifference(K key, K otherKey) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[] otherKeyByte = getKeySerializer().serialize(otherKey);
        Set<byte[]> intersectValues = getConnection().sdiff(keyByte, otherKeyByte);
        if(intersectValues.isEmpty()){
            return null;
        }
        Set<V> set = new HashSet<>();
        for(byte[] valueByte : intersectValues){
            set.add((V)getValueSerializer().deserialize(valueByte));
        }
        return set;
    }

    @Override
    public Long sDifferenceAndStore(K key, K otherKey, K destKey) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[] otherKeyByte = getKeySerializer().serialize(otherKey);
        byte[] destKeyByte = getKeySerializer().serialize(destKey);
        return getConnection().sunionstore(destKeyByte, keyByte, otherKeyByte);
    }

    @Override
    public <V> Set<V> sMembers(K key) {
        byte[] keyByte = getKeySerializer().serialize(key);
        Set<byte[]> intersectValues = getConnection().smembers(keyByte);
        if(intersectValues.isEmpty()){
            return null;
        }
        Set<V> set = new HashSet<>();
        for(byte[] valueByte : intersectValues){
            set.add((V)getValueSerializer().deserialize(valueByte));
        }
        return set;
    }

    @Override
    public Boolean zAdd(K key, V value, double score) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[] valueByte = getValueSerializer().serialize(value);
        Long result =  getConnection().zadd(keyByte, score, valueByte);
        return result == 1;
    }

    @Override
    public Long zRemove(K key, V... values) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[][] valueBytes = new byte[values.length][];
        int count = 0;
        for(Object value : values){
            valueBytes[count++] = getValueSerializer().serialize(value);
        }
        return getConnection().zrem(keyByte, valueBytes);
    }

    @Override
    public Double zIncrementScore(K key, V value, double score) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[] valueByte = getValueSerializer().serialize(value);
        return getConnection().zincrby(keyByte, score, valueByte);
    }

    @Override
    public Long zRank(K key, V value) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[] valueByte = getValueSerializer().serialize(value);
        return getConnection().zrank(keyByte, valueByte);
    }

    @Override
    public Long zReverseRank(K key, V value) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[] valueByte = getValueSerializer().serialize(value);
        return getConnection().zrevrank(keyByte, valueByte);
    }

    @Override
    public <V> Set<V> zRange(K key, long start, long end) {
        byte[] keyByte = getKeySerializer().serialize(key);
        List<byte[]> intersectValues = getConnection().zrange(keyByte, start, end);
        if(intersectValues.isEmpty()){
            return null;
        }
        Set<V> set = new HashSet<>();
        for(byte[] valueByte : intersectValues){
            set.add((V)getValueSerializer().deserialize(valueByte));
        }
        return set;
    }

    @Override
    public <V> Set<V> zReverseRange(K key, long start, long end) {
        byte[] keyByte = getKeySerializer().serialize(key);
        List<byte[]> intersectValues = getConnection().zrevrange(keyByte, start, end);
        if(intersectValues.isEmpty()){
            return null;
        }
        Set<V> set = new HashSet<>();
        for(byte[] valueByte : intersectValues){
            set.add((V)getValueSerializer().deserialize(valueByte));
        }
        return set;
    }

    @Override
    public <V> Set<V> zRangeByScore(K key, double min, double max) {
        byte[] keyByte = getKeySerializer().serialize(key);
        List<byte[]> intersectValues = getConnection().zrangebyscore(keyByte, min, max);
        if(intersectValues.isEmpty()){
            return null;
        }
        Set<V> set = new HashSet<>();
        for(byte[] valueByte : intersectValues){
            set.add((V)getValueSerializer().deserialize(valueByte));
        }
        return set;
    }



    @Override
    public Double zScore(K key, V value) {
        byte[] keyByte = getKeySerializer().serialize(key);
        byte[] valueByte = getValueSerializer().serialize(value);
        return getConnection().zscore(keyByte, valueByte);
    }

    @Override
    public Long zRemoveRange(K key, long start, long end) {
        byte[] keyByte = getKeySerializer().serialize(key);
        return getConnection().zremrangebyrank(keyByte, start, end);
    }

    @Override
    public Long zRemoveRangeByScore(K key, double min, double max) {
        byte[] keyByte = getKeySerializer().serialize(key);
        return getConnection().zremrangebyscore(keyByte, min, max);
    }



    public RedisSerializer getKeySerializer() {
        return keySerializer;
    }

    public void setKeySerializer(RedisSerializer keySerializer) {
        this.keySerializer = keySerializer;
    }

    public RedisSerializer getValueSerializer() {
        return valueSerializer;
    }

    public void setValueSerializer(RedisSerializer valueSerializer) {
        this.valueSerializer = valueSerializer;
    }

    public RedisSerializer getHashKeySerializer() {
        return hashKeySerializer;
    }

    public void setHashKeySerializer(RedisSerializer hashKeySerializer) {
        this.hashKeySerializer = hashKeySerializer;
    }
}